package com.my.db.dataCleaning;

import java.io.*;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.sql.DataSource;

import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 地址数据清洗与关联更新服务类
 * 功能：将 table1 的 address 字段拆分为县、镇、村、组，并根据 table2 名称匹配更新 part_id 和 town_id
 * 特性：分批次处理、批量更新、事务回滚（存储处理数据的文件，回滚sql）、多线程处理、连接池管理
 *
 *  修改某个字段 根据另一张表的结果集
 * 根据表地址 --更新部门id（从部门表中进行匹配）
 * 根据一个表的id name 映射 ，填充主表根据一个字段来匹配的外键
 */
public class DataCleaningAddress {

    // 用于记录日志信息，方便调试和监控程序运行状态
    private static final Logger logger = LoggerFactory.getLogger(DataCleaningAddress.class);

    // 数据库连接参数
    // 数据库连接的 URL，包含了数据库地址、端口、数据库名以及一些连接配置
    private static final String DB_URL = "jdbc:mysql://xxx/yilong?" +
            "useUnicode=true&characterEncoding=utf8" +
            "&autoReconnect=true" +
            "&zeroDateTimeBehavior=CONVERT_TO_NULL" +
            "&transformedBitIsBoolean=true" +
            "&allowPublicKeyRetrieval=true" +
            "&serverTimezone=Asia/Shanghai";
    // 数据库用户名
    private static final String DB_USER = "root";
    // 数据库密码
    private static final String DB_PASSWORD = "";

    // 处理参数配置
    // 每批次处理的数据量，避免一次性处理过多数据导致内存溢出
    private static final int BATCH_SIZE = 999;
    // 线程池的大小，用于多线程处理数据，提高处理效率
    private static final int THREAD_POOL_SIZE = 4;
    // 数据库连接池，用于管理数据库连接，提高连接的复用性和性能
    private static DataSource dataSource;

    // 表和字段配置
    // table1 的表名，存储需要处理的地址数据
    private static final String TABLE1_NAME = "g_parcel";
    // table2 的表名，用于根据地址名称匹配对应的组织信息
    private static final String TABLE2_NAME = "sys_depart";
    // table1 中存储地址信息的字段名
    private static final String TABLE1_ADDRESS_FIELD = "farmer_address";
    // table1 中需要更新的 part_id 字段名
    private static final String TABLE1_PART_ID_FIELD = "depart_id";
    // table1 中需要更新的 town_id 字段名
    private static final String TABLE1_TOWN_ID_FIELD = "town_id";
    // table2 中的主键字段名
    private static final String TABLE2_ID_FIELD = "id";
    // table2 中存储地址名称的字段名
    private static final String TABLE2_NAME_FIELD = "depart_name";
    // table2 中存储上级的字段名
    private static final String TABLE2_PID_FIELD = "parent_id";
    // SQL 语句
    // 查询 table1 中 address 字段不为空且 part_id 字段为空的记录
    private static final String SELECT_ADDRESS_SQL = "SELECT id, " + TABLE1_ADDRESS_FIELD + " FROM " + TABLE1_NAME +
            " WHERE " + TABLE1_ADDRESS_FIELD + " IS NOT NULL AND " + TABLE1_PART_ID_FIELD + " IS NULL";
    // 更新 table1 中的 part_id 和 town_id 字段
    private static final String UPDATE_ORG_ID_SQL = "UPDATE " + TABLE1_NAME +
            " SET " + TABLE1_PART_ID_FIELD + " = ?, " + TABLE1_TOWN_ID_FIELD + " = ? WHERE id = ?";

    /**
     * 某些地址可能直接从镇开始。类似这样“双胜镇永远村四组”，
     */
    // 可配置的默认县名
    private static final String DEFAULT_COUNTY = "仪陇县";
    // 地址拆分正则表达式，支持县名可选
    private static final Pattern ADDRESS_PATTERN = Pattern.compile(
            "(?:([^县]+县)|(" + Pattern.quote(DEFAULT_COUNTY) + "))?" + // 可选县（或默认县）
                    "([^镇]+镇)" + // 镇
                    "([^村]+村)" + // 村
                    "([^组]+组)"); // 组
    //回滚sql
    private static final String ROLLBACK_UPDATE_SQL = "UPDATE " + TABLE1_NAME +
            " SET " + TABLE1_PART_ID_FIELD + " = NULL, " + TABLE1_TOWN_ID_FIELD + " = NULL WHERE id IN (?)";
    // 存储处理数据的文件路径
    private static final String DATA_FILE_PATH = "F:\\临时\\processed_data.txt";

    // 存储 table2 的数据，键为 ID，值为名称
    private static Map<String, String> idNameMap = new HashMap<>();
    // 存储 table2 的数据，键为 ID，值为父级 ID
    private static Map<String, String> idPidMap = new HashMap<>();
    // 存储 table2 的数据，键为完整地址名称，值为组织 ID
    private static Map<String, String> orgFullNameIdMap = new HashMap<>();

    // 静态代码块，在类加载时初始化数据库连接池和读取 table2 数据
    static {
        HikariConfig config = new HikariConfig();
        // 设置数据库连接的 URL
        config.setJdbcUrl(DB_URL);
        // 设置数据库用户名
        config.setUsername(DB_USER);
        // 设置数据库密码
        config.setPassword(DB_PASSWORD);
        // 设置连接池的最大连接数
        config.setMaximumPoolSize(THREAD_POOL_SIZE);
        // 设置连接超时时间（毫秒）
        config.setConnectionTimeout(30000);
        // 设置空闲连接的超时时间（毫秒）
        config.setIdleTimeout(600000);
        // 设置连接的最大存活时间（毫秒）
        config.setMaxLifetime(1800000);
        // 设置连接池的最小空闲连接数
        config.setMinimumIdle(2);
        // 初始化 HikariCP 连接池
        dataSource = new HikariDataSource(config);

        // 读取 table2 数据
        try (Connection conn = dataSource.getConnection();
             PreparedStatement ps = conn.prepareStatement("SELECT " + TABLE2_NAME_FIELD + ", " + TABLE2_ID_FIELD + ", " + TABLE2_PID_FIELD + " FROM " + TABLE2_NAME)) {
            try (ResultSet rs = ps.executeQuery()) {
                while (rs.next()) {
                    String name = rs.getString(TABLE2_NAME_FIELD);
                    String id = rs.getString(TABLE2_ID_FIELD);
                    String pid = rs.getString(TABLE2_PID_FIELD);
                    idNameMap.put(id, name);
                    idPidMap.put(id, pid);
                }
            }
            // 递归构建完整名称并存储到映射中
            for (String id : idNameMap.keySet()) {
                String fullName = buildFullName(id);
                orgFullNameIdMap.put(fullName, id);
            }
        } catch (SQLException e) {
            logger.error("读取 table2 数据失败", e);
        }
    }

    // 递归构建完整名称
    private static String buildFullName(String id) {
        String name = idNameMap.get(id);
        String pid = idPidMap.get(id);
        if (pid == null || pid.isEmpty() || "0".equals(pid)) {
            return name;
        }
        return buildFullName(pid) + name;
    }


    /**
     * 程序入口方法，启动地址清洗和更新操作
     * @param args 命令行参数
     */
    public static void main(String[] args) {
        try {
            // 执行地址清洗和更新操作
            performAddressCleaning();
//            rollbackFromFile();
        } catch (IOException | SQLException | InterruptedException e) {
            // 记录地址清洗失败的日志信息
            logger.error("地址清洗或回滚失败", e);
        }
    }

    /**
     * 从文件读取数据进行回滚
     * @throws IOException 文件操作异常
     * @throws SQLException 数据库操作异常
     */
    public static void rollbackFromFile() throws IOException, SQLException, InterruptedException {
        List<String> allLines = new ArrayList<>();
        BufferedReader reader = new BufferedReader(new FileReader(DATA_FILE_PATH));
        String line;
        while ((line = reader.readLine()) != null) {
            allLines.add(line);
        }
        reader.close();

        int totalBatches = (int) Math.ceil((double) allLines.size() / BATCH_SIZE);
        AtomicInteger totalRolledBack = new AtomicInteger();
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

        for (int i = 0; i < totalBatches; i++) {
            int startIndex = i * BATCH_SIZE;
            int endIndex = Math.min(startIndex + BATCH_SIZE, allLines.size());
            List<String> batch = allLines.subList(startIndex, endIndex);
            executorService.submit(() -> {
                try {
                    if (batch.isEmpty()) return;
                    int rolledBack = processRollbackBatch(batch);
                    totalRolledBack.addAndGet(rolledBack);
                    logger.info("本批次回滚数据量: {}", rolledBack);
                } catch (SQLException e) {
                    logger.error("批次回滚失败，正在尝试重试...", e);
                    // 简单重试逻辑，可根据实际情况调整
                    try {
                        Thread.sleep(1000); // 等待 1 秒后重试
                        int rolledBack = processRollbackBatch(batch);
                        totalRolledBack.addAndGet(rolledBack);
                        logger.info("重试成功，本批次回滚数据量: {}", rolledBack);
                    } catch (SQLException | InterruptedException ex) {
                        logger.error("重试失败，批次回滚数据量未更新", ex);
                    }
                }
            });
        }

        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
                logger.warn("线程池等待超时，已强制关闭");
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
            Thread.currentThread().interrupt();
            logger.error("线程池等待被中断，已强制关闭", e);
        }
        logger.info("数据已从文件回滚，总共回滚数据量: {}", totalRolledBack);
    }

    private static int processRollbackBatch(List<String> batch) throws SQLException {
        if (batch.isEmpty()) {
            logger.info("当前批次数据为空，跳过回滚");
            return 0;
        }
        List<String> ids = new ArrayList<>();
        for (String line : batch) {
            String[] parts = line.split(",");
            String id = parts[0];
            ids.add(id);
        }
        logger.info("本批次回滚的 id 列表: {}", ids);
        try (Connection conn = dataSource.getConnection()) {
            // 开启事务
            conn.setAutoCommit(false);
            try {
                StringBuilder inClause = new StringBuilder("(");
                for (int i = 0; i < ids.size(); i++) {
                    String id = ids.get(i);
                    inClause.append("'").append(id).append("'");
                    if (i < ids.size() - 1) {
                        inClause.append(",");
                    }
                }
                inClause.append(")");
                String sql = ROLLBACK_UPDATE_SQL.replace("(?)", inClause.toString());
                logger.info("执行回滚 SQL: {}", sql);
                try (PreparedStatement ps = conn.prepareStatement(sql)) {
                    int rowsAffected = ps.executeUpdate();
                    logger.info("回滚操作影响的行数: {}", rowsAffected);
                    // 提交事务
                    conn.commit();
                    return rowsAffected;
                }
            } catch (SQLException e) {
                // 发生异常时，回滚事务
                conn.rollback();
                throw e;
            }
        }
    }
    /**
     * 核心处理方法，执行地址清洗和更新操作
     * @throws SQLException 数据库操作异常
     * @throws InterruptedException 线程中断异常
     */
    public static void performAddressCleaning() throws SQLException, InterruptedException, IOException {
        // 创建一个固定大小的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
        // 获取所有需要处理的地址记录
        List<AddressRecord> addressRecords = getAllAddressRecords();
        // 计算总共需要处理的批次数量
        int totalBatches = (int) Math.ceil((double) addressRecords.size() / BATCH_SIZE);
        // 用于统计总共处理的数据量，使用 AtomicInteger 保证线程安全
        AtomicInteger totalProcessed = new AtomicInteger();
        BufferedWriter writer = new BufferedWriter(new FileWriter(DATA_FILE_PATH));

        // 分批次提交任务到线程池进行处理
        for (int i = 0; i < totalBatches; i++) {
            int startIndex = i * BATCH_SIZE;
            int endIndex = Math.min(startIndex + BATCH_SIZE, addressRecords.size());
            List<AddressRecord> batch = addressRecords.subList(startIndex, endIndex);
            executorService.submit(() -> {
                try {
                    if (batch.isEmpty()) return;
                    // 处理当前批次的数据
                    int processed = processBatch(batch, writer);
                    // 累加处理的数据量
                    totalProcessed.addAndGet(processed);
                    // 记录本批次处理的数据量
                    logger.info("本批次处理数据量: {}", processed);
                } catch (SQLException | IOException e) {
                    logger.error("批次处理失败，正在尝试重试...", e);
                    // 简单重试逻辑，可根据实际情况调整
                    try {
                        Thread.sleep(1000); // 等待 1 秒后重试
                        int processed = processBatch(batch, writer);
                        totalProcessed.addAndGet(processed);
                        logger.info("重试成功，本批次处理数据量: {}", processed);
                    } catch (SQLException | IOException | InterruptedException ex) {
                        logger.error("重试失败，本批次处理数据量未更新", ex);
                    }
                }
            });
        }

        // 关闭线程池，不再接受新的任务
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
                logger.warn("线程池等待超时，已强制关闭");
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
            Thread.currentThread().interrupt();
            logger.error("线程池等待被中断，已强制关闭", e);
        }
        writer.close();
        // 记录地址清洗完成的日志信息，包含总共处理的数据量
        logger.info("地址清洗完成，总共处理数据量: {}", totalProcessed);
    }

    /**
     * 获取所有需要处理的地址记录
     * @return 地址记录列表
     * @throws SQLException 数据库操作异常
     */
    private static List<AddressRecord> getAllAddressRecords() throws SQLException {
        List<AddressRecord> records = new ArrayList<>();
        try (
                // 从连接池获取数据库连接
                Connection conn = dataSource.getConnection();
                // 预编译 SQL 语句，提高性能
                PreparedStatement ps = conn.prepareStatement(SELECT_ADDRESS_SQL)
        ) {
            try (
                    // 执行 SQL 查询，获取结果集
                    ResultSet rs = ps.executeQuery()
            ) {
                // 遍历结果集，将每条记录封装成 AddressRecord 对象并添加到列表中
                while (rs.next()) {
                    String address = rs.getString(TABLE1_ADDRESS_FIELD);
                    String id = rs.getString("id");
                    records.add(new AddressRecord(id, address));
                }
            }
        }
        return records;
    }

    /**
     * 处理单个批次的数据
     * @param batch 当前批次的地址记录列表
     * @return 本批次处理的数据量
     * @throws SQLException 数据库操作异常
     */
    private static int processBatch(List<AddressRecord> batch, BufferedWriter writer) throws SQLException, IOException {
        // 如果批次为空，直接返回 0
        if (batch.isEmpty()) return 0;

        try (
                // 从连接池获取数据库连接
                Connection conn = dataSource.getConnection()
        ) {
            // 开启事务，确保数据操作的一致性
            conn.setAutoCommit(false);
            try {
                // 根据当前批次的地址记录，获取对应的组织信息映射
                Map<String, OrgInfo> orgMap = getOrgMap(batch);
                // 更新 table1 中的 part_id 和 town_id 字段
                int[] updateCounts = updateOrgInfo(conn, orgMap, writer);
                // 提交事务
                conn.commit();
                // 计算本批次处理的数据量
                return Arrays.stream(updateCounts).sum();
            } catch (SQLException | IOException e) {
                // 发生异常时，回滚事务
                conn.rollback();
                throw e;
            }
        }
    }

    /**
     * 根据地址记录列表，获取对应的组织信息映射
     * @param batch 地址记录列表
     * @return 组织信息映射，键为记录 ID，值为组织信息对象
     */
    private static Map<String, OrgInfo> getOrgMap(List<AddressRecord> batch) {
        Map<String, OrgInfo> orgMap = new HashMap<>();
        for (AddressRecord record : batch) {
            // 拆分地址字符串为县、镇、村、组
            String[] parts = splitAddress(record.getAddress());
            if (parts == null) continue;

            String county = parts[0];
            String town = parts[1];
            String village = parts[2];
            String group = parts[3];

            String matchedId = null;
            String townIdValue = null;

            // 组合不同的匹配名称
            String fullGroupMatch = county + town + village + group;
            String groupMatch = town + village + group;
            String fullVillageMatch = county + town + village;
            String villageMatch = town + village;
            String fullTownMatch = county + town;

            // 先尝试全匹配（县 + 镇 + 村 + 组）
            matchedId = orgFullNameIdMap.get(fullGroupMatch);
            if (matchedId != null) {
                townIdValue = orgFullNameIdMap.get(fullTownMatch);
            }

            // 若全匹配失败，尝试 镇 + 村 + 组 匹配
            if (matchedId == null) {
                matchedId = orgFullNameIdMap.get(groupMatch);
                if (matchedId != null) {
                    townIdValue = orgFullNameIdMap.get(town);
                }
            }

            // 若 镇 + 村 + 组 匹配失败，尝试 县 + 镇 + 村 匹配
            if (matchedId == null) {
                matchedId = orgFullNameIdMap.get(fullVillageMatch);
                if (matchedId != null) {
                    townIdValue = orgFullNameIdMap.get(fullTownMatch);
                }
            }

            // 若 县 + 镇 + 村 匹配失败，尝试 镇 + 村 匹配
            if (matchedId == null) {
                matchedId = orgFullNameIdMap.get(villageMatch);
                if (matchedId != null) {
                    townIdValue = orgFullNameIdMap.get(town);
                }
            }

            // 若 镇 + 村 匹配失败，尝试 县 + 镇 匹配
            if (matchedId == null) {
                matchedId = orgFullNameIdMap.get(fullTownMatch);
                if (matchedId != null) {
                    townIdValue = matchedId;
                }
            }

            // 若 县 + 镇 匹配失败，尝试 镇 匹配
            if (matchedId == null) {
                matchedId = orgFullNameIdMap.get(town);
                if (matchedId != null) {
                    townIdValue = matchedId;
                }
            }

            // 若 镇 匹配失败，尝试 县 匹配
            if (matchedId == null) {
                matchedId = orgFullNameIdMap.get(county);
                if (matchedId != null) {
                    townIdValue = matchedId;
                }
            }

            if (matchedId != null) {
                orgMap.put(record.getId(), new OrgInfo(matchedId, townIdValue));
            }
        }
        return orgMap;
    }
    /**
     * 更新 table1 中的 part_id 和 town_id 字段
     * @param conn 数据库连接
     * @param orgMap 组织信息映射
     * @return 每个更新操作影响的行数数组
     * @throws SQLException 数据库操作异常
     */
    private static int[] updateOrgInfo(Connection conn, Map<String, OrgInfo> orgMap, BufferedWriter writer) throws SQLException, IOException {
        try (
                // 预编译 SQL 语句，用于更新 table1 中的 part_id 和 town_id 字段
                PreparedStatement ps = conn.prepareStatement(UPDATE_ORG_ID_SQL)
        ) {
            // 遍历组织信息映射
            for (Map.Entry<String, OrgInfo> entry : orgMap.entrySet()) {
                // 设置 SQL 语句的参数
                ps.setString(1, entry.getValue().getOrgId());
                ps.setString(2, entry.getValue().getTownId());
                ps.setString(3, entry.getKey());
                // 将更新操作添加到批处理中
                ps.addBatch();
                synchronized (writer) {
                    // 将处理的数据写入文件
                    writer.write(entry.getKey() + "," + entry.getValue().getOrgId() + "," + entry.getValue().getTownId());
                    writer.newLine();
                }
            }
            // 执行批处理更新操作
            return ps.executeBatch();
        }
    }

    /**
     * 拆分地址字符串为县、镇、村、组
     * @param address 地址字符串
     * @return 包含县、镇、村、组的字符串数组，如果格式不正确返回 null
     */
    private static String[] splitAddress(String address) {
        Matcher matcher = ADDRESS_PATTERN.matcher(address);
        if (matcher.find()) {
            String county = matcher.group(1) != null ? matcher.group(1) : DEFAULT_COUNTY;
            String town = matcher.group(3);
            String village = matcher.group(4);
            String group = matcher.group(5);
            return new String[]{county, town, village, group};
        }
        logger.warn("地址格式不正确: {}", address);
        return null;
    }

    /**
     * 内部类，用于封装地址记录的信息
     */
    private static class AddressRecord {
        // 记录的 ID
        private final String id;
        // 地址信息
        private final String address;

        /**
         * 构造方法，初始化记录的 ID 和地址信息
         * @param id 记录的 ID
         * @param address 地址信息
         */
        public AddressRecord(String id, String address) {
            this.id = id;
            this.address = address;
        }

        /**
         * 获取记录的 ID
         * @return 记录的 ID
         */
        public String getId() {
            return id;
        }

        /**
         * 获取地址信息
         * @return 地址信息
         */
        public String getAddress() {
            return address;
        }
    }

    /**
     * 内部类，用于封装组织信息
     */
    @AllArgsConstructor
    @Data
    private static class OrgInfo {
        // 组织的 ID
        private final String orgId;
        private final String townId;
    }
}